package com.dianping.abel.thread;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.util.*;
import java.util.concurrent.*;

/**
 * 线程池工厂类:提供不同任务类型适用的线程池
 * @author xiachao
 * @date 2019/2/12
 */
@Slf4j
public class ThreadPoolUtil {
    /**
     * 当前机器cpu核心数
     */
    private static final int PROCCESSOR_NUM = Runtime.getRuntime().availableProcessors();

    private static final int WAIT_QUEUE_SIZE = 2000;

    private static final long DEAULT_KEEP_ALIVE_TIME = 60L;

    private static final long MONITOR_REPORT_CYCLE = 10L;

    private static volatile Map<String, ExecutorService> poolContainer = new ConcurrentHashMap<>();


    public static ThreadPoolExecutor getDefaultThreadPool(int coreSize, int maxSize){
        return new ThreadPoolExecutor(coreSize, maxSize, DEAULT_KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue<>(WAIT_QUEUE_SIZE));
    }

    /**
     * 获取IO密集任务类型线程池
     * @return
     */
    public static ThreadPoolExecutor getThreadPoolForIO(){
        return ThreadPoolUtil.getDefaultThreadPool(PROCCESSOR_NUM + 1, PROCCESSOR_NUM * 2);
    }

    /**
     * 获取CPU计算型任务类线程池
     * @return
     */
    public static ThreadPoolExecutor getThreadPoolForCPU(){
        return ThreadPoolUtil.getDefaultThreadPool(PROCCESSOR_NUM * 2 + 1, PROCCESSOR_NUM * 5);
    }

    /**
     * 处理提交的异步任务并返回结果集合
     * @param futures   任务单元
     * @param timeout   处理超时时间
     * @param timeUnit  时间单元
     * @param <T>
     * @return
     */
    public static <T> List<T> fetchAyncResults(List<Future<T>> futures, int timeout, TimeUnit timeUnit){
        Iterator<Future<T>> iterator = futures.iterator();
        List<T> result = new ArrayList<>();
        while(iterator.hasNext()){
            try {
                T data = iterator.next().get(timeout, timeUnit);
                result.add(data);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("FetchAyncResultFailed", e);
            }
        }
        return result;
    }

    public static <T> List<T> fetchAyncResults(List<Future<T>> futures){
        return ThreadPoolUtil.fetchAyncResults(futures, 0, TimeUnit.SECONDS);
    }

    public static <T> List<T> fetchAyncResults(List<Future<T>> futures, int timeout){
        return ThreadPoolUtil.fetchAyncResults(futures, timeout, TimeUnit.SECONDS);
    }

    /**
     * 对线程池状态进行监控
     * @param monitor
     * @param monicker
     * @param poolExecutor
     */
    private void doMonitor(boolean monitor, String monicker, ThreadPoolExecutor poolExecutor){
        if(monitor == false || StringUtils.isBlank(monicker)){
            return;
        }
        Runnable monitorTask = ()->{
            while(true){
                log.info(monicker+" : coreSize -> "+poolExecutor.getCorePoolSize()+" MaximunPoolSize -> "+poolExecutor.getMaximumPoolSize()+" ActiveCount -> "+poolExecutor.getActiveCount());
                try{
                    Thread.sleep(MONITOR_REPORT_CYCLE * 1000);
                }catch (Exception e){
                    //do nothing
                }
            }
        };
        new Thread(monitorTask).start();
    }
}
